package com.fly.http;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.lang3.RandomUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.http.MediaType;
import org.springframework.web.reactive.function.client.WebClient;

import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Mono;

/**
 * WebClient是RestTemplete的替代品，有更好的响应式能力，支持异步调用<br>
 * 
 * https://blog.csdn.net/zzhongcy/article/details/105412842
 * 
 */
@Slf4j
public class FluxWebClient
{
    private List<String> urls = new ArrayList<>();
    
    // 缓冲区默认256k，设为-1以解决报错Exceeded limit on max bytes to buffer : 262144
    private WebClient webClient = WebClient.builder().codecs(configurer -> configurer.defaultCodecs().maxInMemorySize(-1)).build();
    
    public void visitAll()
    {
        // block转换为同步调用
        if (urls.isEmpty())
        {
            log.info("★★★★★★★★ urls isEmpty, now get urls from api ★★★★★★★★");
            Mono<String> mono = webClient.get().uri("https://00fly.online/upload/urls.txt").acceptCharset(StandardCharsets.UTF_8).accept(MediaType.TEXT_HTML).retrieve().bodyToMono(String.class);
            urls = Arrays.asList(StringUtils.split(mono.block(), "\r\n"));
        }
        
        // 异步访问
        AtomicInteger count = new AtomicInteger(0);
        urls.stream()
            .filter(url -> RandomUtils.nextBoolean())
            .forEach(url -> webClient.get()
                .uri(url)
                .acceptCharset(StandardCharsets.UTF_8)
                .accept(MediaType.TEXT_HTML)
                .retrieve()
                .bodyToMono(String.class)
                .subscribe(r -> log.info("process complted: {}. {}", count.incrementAndGet(), url), e -> log.error(e.getMessage())));
        log.info("total:{} ==> ############## 异步请求已提交 ##############", urls.size());
    }
}
